package com.iphanta.community.activity.voting.influxdb;

import com.iphanta.community.activity.voting.entity.TimeRange;
import com.iphanta.community.common.BusinessException;
import com.iphanta.community.common.utils.ResultUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.springframework.util.CollectionUtils;

import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;

/**
 * @author 墨声 E-mail: zehong.hongframe.huang@gmail.com
 * create time: 4/11/22 11:23 AM
 */
public class VoteNumFlux {

    private static final String MEASUREMENT = "activity_vote_record";

    private static final String BUCKET = InfluxdbUtils.bucket;

    public static final String DEFAULT_RANGE = "  |> range(start: -30d, stop: now())\n";

    public static String rangeRanking(String activityId, TimeRange timeRange, List<Integer> voteTypes) {
        return rangeTopN(activityId, timeRange, null, voteTypes);

    }

    public static String rangeTopN(String activityId, TimeRange timeRange, Integer topN, List<Integer> voteTypes) {
        String range = "";
        if (Objects.nonNull(timeRange)) {
            range =
                    "  |> range(start: %s, stop: %s)\n";
            range = String.format(range,
                    timeRange.getStart(),
                    timeRange.getStop());
        }
        return customRange(range, topN, activityId, voteTypes);
    }

    public static String recently(String activityId, Integer recently, List<Integer> voteTypes) {
        return recentlyTopN(activityId, recently, null, voteTypes);
    }

    public static String recentlyTopN(String activityId, Integer recently, Integer topN, List<Integer> voteTypes) {
        String range = "";
        if (Objects.nonNull(recently)) {
            range =
                    "  |> range(start: -%smo, stop: now())\n";
            range = String.format(range,
                    recently);
        }
        return customRange(range, topN, activityId, voteTypes);
    }


    public static String customRange(String range, Integer topN, String activityId, List<Integer> voteTypes) {
        StringBuffer flux = new StringBuffer();
        String from =
                "from(bucket: \"%s\")\n";
        from = String.format(from, BUCKET);
        flux.append(from);

        if (StringUtils.isNotBlank(range)) {
            flux.append(range);
        } else {
            flux.append(DEFAULT_RANGE);
        }

        String filter =
                "  |> filter(fn: (r) => r[\"_measurement\"] == \"%s\")\n" +
                        "  |> filter(fn: (r) => r[\"_field\"] == \"vote_number\")\n" +
                        "  |> filter(fn: (r) => r[\"activity_id\"] == \"%s\")\n";
        if (!CollectionUtils.isEmpty(voteTypes)) {
            List<String> voteTypesFilter = new ArrayList<>();
            for (Integer voteType : voteTypes) {
                voteTypesFilter.add(String.format("r[\"vote_type\"] == \"%s\"", voteType));
            }
            String s = "  |> filter(fn: (r) => " + String.join(" or ", voteTypesFilter) + ")\n";
            filter = filter + s;
        }
        filter = String.format(filter, MEASUREMENT, activityId);
        flux.append(filter);

        String aggregate =
                "  |> group(columns: [\"activity_id\", \"dynamic_id\"], mode: \"by\")\n";
        String reduce = "  |> reduce(fn: (r, accumulator) => \n" +
                "    ({ vote_sum: r._value + accumulator.vote_sum, vote_count: accumulator.vote_count + 1}),\n" +
                "    identity: {vote_sum: 0.0, vote_count : 0})\n";
        flux.append(aggregate).append(reduce);

//        topPecentile();

        if (Objects.nonNull(topN) && topN > 0) {
            String highestCurrent = "  |> highestMax(column: \"vote_sum\", n: %s, groupColumns: [\"dynamic_id\"])\n";
            flux.append(String.format(highestCurrent, topN));
        }
        String yield =
                "  |> yield(name : \"result_1\")";
        flux.append(yield);
        return flux.toString();
    }

    public static String topPecentile(Integer segmentalNum, Integer compression) {
        String quantile = "  |> group(columns: [\"activity_id\"], mode: \"by\")\n" +
                "  |> quantile(q: %s, method: \"estimate_tdigest\", compression: %s)\n";
        if (segmentalNum > 100 || segmentalNum < 1) {
            throw new BusinessException("等分位数不得小于0，或大于100", ResultUtils.FAIL_ON_BUSINESS.result());
        }
        Double segment = 1.0 / segmentalNum;
        do {

        } while (segment <= 1.0);
        return String.format(quantile, segmentalNum, compression);
    }

}
